/*
 * Copyright © 2018 www.noark.xyz All Rights Reserved.
 *
 * 感谢您选择Noark框架，希望我们的努力能为您提供一个简单、易用、稳定的服务器端框架 ！
 * 除非符合Noark许可协议，否则不得使用该文件，您可以下载许可协议文件：
 *
 *        http://www.noark.xyz/LICENSE
 *
 * 1.未经许可，任何公司及个人不得以任何方式或理由对本框架进行修改、使用和传播;
 * 2.禁止在本项目或任何子项目的基础上发展任何派生版本、修改版本或第三方版本;
 * 3.无论你对源代码做出任何修改和改进，版权都归Noark研发团队所有，我们保留所有权利;
 * 4.凡侵犯Noark版权等知识产权的，必依法追究其法律责任，特此郑重法律声明！
 */
package xyz.noark.network.rpc.shuttle;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import xyz.noark.network.codec.AbstractPacketCodec;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.*;

import static xyz.noark.log.LogHelper.logger;

/**
 * 连接工厂类，主要负责处理连接事项
 *
 * @author 小流氓[176543888@qq.com]
 * @since 4.0
 */
class ShuttleConnectionFactory {
    // 全局 EventLoopGroup，复用线程资源
    private final EventLoopGroup group = new NioEventLoopGroup();
    private final Bootstrap bootstrap;
    // 记录每个目标服务器的连接状态: key = "host:port"
    private final Map<Integer, TargetConnectionState> connectionStates = new ConcurrentHashMap<>();
    // 调度重连任务
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
    private final ShuttleConnectionManager connectionManager;
    private final ShuttleConnectionHandler rpcConnectorHandler;

    ShuttleConnectionFactory(ShuttleConnectionManager connectionManager, AbstractPacketCodec packetCodec) {
        this.connectionManager = connectionManager;
        this.rpcConnectorHandler = new ShuttleConnectionHandler();
        this.bootstrap = new Bootstrap();
        this.bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) {
                ch.pipeline().addLast("decoder", packetCodec.lengthDecoder());
                ch.pipeline().addLast("encoder", packetCodec.lengthEncoder());
                ch.pipeline().addLast(rpcConnectorHandler);
            }
        });
    }

    /**
     * 连接到指定的目标服务器（host:port），如果已连接则忽略
     */
    public synchronized void connect(Integer serverUid, String host, int port) {
        TargetConnectionState state = connectionStates.computeIfAbsent(serverUid, k -> new TargetConnectionState(serverUid, host, port));
        if (state.isConnectingOrConnected()) {
            logger.warn("正在连接或已连接到目标，跳过重复请求: {}", serverUid);
            return;
        }
        state.markConnecting();
        doConnect(state);
    }

    /**
     * 主动断开某个目标连接（可选）
     */
    public void disconnect(Integer serverUid) {
        TargetConnectionState state = connectionStates.get(serverUid);
        if (state != null) {
            state.cancelReconnect();
            state.markDisconnected();
            Channel channel = state.getChannel();
            if (channel != null && channel.isActive()) {
                channel.close();
            }
            logger.info("连接已断开 serverUid={}, host={}, port={}", serverUid, state.getHost(), state.getPort());
        }
    }

    /**
     * 断开所有连接
     */
    public void disconnectAll() {
        connectionStates.values().forEach(this::disconnectState);
    }

    private void disconnectState(TargetConnectionState state) {
        state.cancelReconnect();
        state.markDisconnected();
        Channel channel = state.getChannel();
        if (channel != null && channel.isActive()) {
            channel.close();
        }
    }

    private void doConnect(TargetConnectionState state) {
        InetSocketAddress address = new InetSocketAddress(state.getHost(), state.getPort());
        ChannelFuture future = bootstrap.connect(address);
        future.addListener((ChannelFutureListener) f -> {
            if (f.isSuccess()) {
                logger.info("成功连接到目标服务器: serverUid={}, address={}", state.getServerUid(), address);
                state.setChannel(f.channel());
                state.markConnected();
                state.cancelReconnect();
                // 停止重试
                // 注册到连接管理器（outbound=true）
                connectionManager.register(state.getServerUid(), f.channel());
                // 添加监听：连接断开后自动触发重连
                f.channel().closeFuture().addListener((ChannelFutureListener) closeFuture -> {
                    logger.warn("与目标服务器断开，准备重连: serverUid={}, address={}", state.getServerUid(), address);
                    state.markDisconnected();
                    scheduleReconnect(state);
                });
            } else {
                logger.warn("重连失败: serverUid={}, address={}, {}秒后重试...", state.getServerUid(), address, 3);
                scheduleReconnect(state);
            }
        });
    }

    private void scheduleReconnect(TargetConnectionState state) {
        state.scheduleReconnect(() -> {
            if (state.isConnectingOrConnected()) return;
            // 防止并发重连
            doConnect(state);
        });
    }

    /**
     * 关闭工厂资源
     */
    public void shutdown() {
        disconnectAll();
        scheduler.shutdown();
        group.shutdownGracefully();
    }

    // 内部类：记录每个目标的连接状态
    private class TargetConnectionState {
        private final Integer serverUid;
        private final String host;
        private final int port;
        private volatile Channel channel;
        private volatile boolean connectingOrConnected = false;
        private ScheduledFuture<?> reconnectTask;

        public TargetConnectionState(Integer serverUid, String host, int port) {
            this.serverUid = serverUid;
            this.host = host;
            this.port = port;
        }

        public Integer getServerUid() {
            return serverUid;
        }

        public String getHost() {
            return host;
        }

        public int getPort() {
            return port;
        }

        public Channel getChannel() {
            return channel;
        }

        public synchronized void setChannel(Channel channel) {
            this.channel = channel;
        }

        public synchronized boolean isConnectingOrConnected() {
            return connectingOrConnected;
        }

        public synchronized void markConnecting() {
            this.connectingOrConnected = true;
        }

        public synchronized void markConnected() {
            this.connectingOrConnected = true;
        }

        public synchronized void markDisconnected() {
            this.connectingOrConnected = false;
            this.channel = null;
        }

        public synchronized void scheduleReconnect(Runnable task) {
            cancelReconnect();
            this.reconnectTask = scheduler.schedule(task, 3, TimeUnit.SECONDS);
        }

        public synchronized void cancelReconnect() {
            if (reconnectTask != null && !reconnectTask.isCancelled()) {
                reconnectTask.cancel(false);
                reconnectTask = null;
            }
        }
    }
}